package com.queue.cli.handler;

import com.alibaba.fastjson.JSON;
import com.queue.cli.client.Consumer;
import com.queue.cli.constant.MessageType;
import com.queue.cli.listener.MessageData;
import com.queue.cli.listener.MessageListener;
import com.queue.common.message.Message;
import com.queue.common.protocol.MessageUtils;
import com.queue.common.protocol.Protocol;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.StandardCharsets;

public class ConsumerHandler extends SimpleChannelInboundHandler<Protocol> {
    /**
     * 要消费的队列满
     */
    private String destination;
    /**
     * 监听器
     */
    private MessageListener messageListener;
    /**
     * 自动签名(默认开启)
     */
    private boolean isAutoAcknowledge = true;

    public ConsumerHandler(String destination){
        this.destination = destination;
    }

    /**
     * 消息监听通道，假如有消息就会调用这个方法
     * @param ctx
     * @param protocol
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Protocol protocol) throws Exception {
        //将字节数组转换成消息对象
        Message message = MessageUtils.reverse(protocol.getContent());
        if (message!=null){
            execListener(new MessageData(this,message.getContent()));
            if (isAutoAcknowledge){
                ack(ctx.channel(), message.getMessageId(), message.getDestination(),message.getDestinationType());
            }
        }
    }

    /**
     * 执行监听事件
     * @param data
     */
    private void execListener(MessageData data) {
        this.messageListener.getMessage(data);
    }

    /**
     * 向服务端签收指定的消息
     * @param channel
     * @param messageId
     * @param destination
     * @param destinationType
     */
    private void ack(Channel channel, Long messageId, String destination, int destinationType) {
        Message message = new Message(messageId, MessageType.ACK.getType(), null,destination, destinationType,false,0,null);
        byte[] content = JSON.toJSONString(message).getBytes(StandardCharsets.UTF_8);
        Protocol protocol = new Protocol(content.length,content);
        channel.writeAndFlush(protocol);
    }


    /**
     * 获取消息监听器
     * @return
     */
    public MessageListener getMessageListener() {
        return this.messageListener;
    }

    public void setAutoAcknowledge(boolean autoAcknowledge) {
        this.isAutoAcknowledge = autoAcknowledge;
    }

    /**
     * 消息监听器
     * @param listener
     */
    public void createListener(MessageListener listener) {
        this.messageListener = listener;
    }
}
